package com.myorg.util.email;

import java.io.File;
import java.io.IOException;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.myorg.domain.EmailDelimiter;
import com.myorg.util.CompressionHelper;
import com.myorg.util.HdfsHelper;
import com.myorg.util.Utility;
import com.myorg.util.email.pff.PSTMessage;

public class EmailLoaderPst2Hdfs {
	final static Logger logger = LoggerFactory.getLogger(EmailLoaderPst2Hdfs.class);

	public static void savePst2Hdfs(final File pstSrcFile, String targetPathInHdfs, final boolean compress) {
		HdfsHelper helper = new HdfsHelper();
		helper.saveToHDFS(targetPathInHdfs, new HdfsHelper.IHDFSFileSaver() {
			@Override
			public void save(final FSDataOutputStream targetOutputStream) {
				OutlookStorageTableReader.parse(pstSrcFile, new OutlookStorageTableReader.IOutlookEmailProcessor() {
					@Override
					public void process(PSTMessage msg) throws IOException {
						if (msg != null) {
							String body = msg.getBody();
							if (body != null) {
								if (logger.isDebugEnabled()) {
									logger.debug("\n" + body + "\n");
								}
								EmailDelimiter del = new EmailDelimiter(Utility.generateUUID(), msg
										.getMessageDeliveryTime());
								if (compress) {
									CompressionHelper.compressAndSave(body, del.toString(), targetOutputStream);
								} else {
									targetOutputStream.writeBytes(body + "\n");
									targetOutputStream.writeBytes(del.toString() + "\n");
								}
							}
						}
					}
				});
			}
		});
	}
}
